package cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class JOB07_各品牌销售额TOPN商品 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射订单主表
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            BIGINT ,          " +
                        " status        INT ,             " +
                        " modify_time   timestamp(3)  ,   " +
                        " rt as modify_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED                 " +
                        " ) WITH (                                      " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );


        // 建表，映射订单详情表
        tenv.executeSql(
                "CREATE TABLE order_item_mysql (    " +
                        " id                BIGINT ,         " +
                        " order_id          BIGINT ,         " +
                        " product_id        BIGINT ,         " +
                        " product_name      string  ,        " +
                        " product_brand     string  ,        " +
                        " real_amount       decimal(10,2),   " +
                        " primary key (id) not enforced      " +
                        " ) WITH (                                     " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order_item'          " +
                        ")"
        );


        // 关联
        tenv.executeSql(
                " create temporary view joined_view as  "+
                        " select                                "+
                        "     o.id as order_id,                 "+
                        "     o.status as order_status,         "+
                        "     o.modify_time,                    "+
                        "     o.rt,                             "+
                        "     i.product_id,                     "+
                        "     i.product_name,                   "+
                        "     i.product_brand,                  "+
                        "     i.real_amount                     "+
                        " from order_mysql o                    "+
                        " join order_item_mysql i               "+
                        " on o.id=i.order_id  and o.status>0    "

        );


        // 表转流
        Table table = tenv.from("joined_view");
        DataStream<Row> changelogStream = tenv.toChangelogStream(table);

        // 流转表
        Table table2 = tenv.fromChangelogStream(changelogStream,
                Schema.newBuilder()
                        .column("order_id", DataTypes.BIGINT())
                        .column("order_status", DataTypes.INT())
                        .column("modify_time", DataTypes.TIMESTAMP(3))
                        .column("rt", DataTypes.TIMESTAMP(3))
                        .column("product_id", DataTypes.BIGINT())
                        .column("product_name", DataTypes.STRING())
                        .column("product_brand", DataTypes.STRING())
                        .column("real_amount", DataTypes.DECIMAL(10, 2))
                        .columnByExpression("rt2", "modify_time")
                        .watermark("rt2", "rt2 - interval '0' second ")
                        .build());
        tenv.createTemporaryView("tmp2",table2);


        // 统计sql
        tenv.executeSql(
                "with o1 AS (                                                                                                       "+
                "    SELECT                                                                                                                  "+
                        "        TUMBLE_START(rt2,interval '24' hour) as window_start,                                                       "+
                        "        TUMBLE_END(rt2,interval '24' hour) as window_end,                                                           "+
                        "    	product_brand,                                                                                               "+
                        "    	product_name,                                                                                                "+
                        "    	product_id,                                                                                                  "+
                        "    	SUM(real_amount) as real_amount                                                                              "+
                        "    FROM tmp2                                                                                                       "+
                        "    GROUP BY                                                                                                        "+
                        "        TUMBLE(rt2,interval '24' hour),                                                                             "+
                        "    	product_brand,                                                                                               "+
                        "    	product_name,                                                                                                "+
                        "    	product_id                                                                                                   "+
                        ")	                                                                                                                 "+
                        "                                                                                                                    "+
                        "SELECT                                                                                                              "+
                        "    window_start,                                                                                                   "+
                        "    window_end,                                                                                                     "+
                        "    product_brand,                                                                                                  "+
                        "    product_name,                                                                                                   "+
                        "    product_id,                                                                                                     "+
                        "    real_amount,                                                                                                    "+
                        "	rn                                                                                                               "+
                        "FROM (                                                                                                              "+
                        "SELECT                                                                                                              "+
                        "    window_start,                                                                                                   "+
                        "    window_end,                                                                                                     "+
                        "    product_brand,                                                                                                  "+
                        "    product_name,                                                                                                   "+
                        "    product_id,                                                                                                     "+
                        "    real_amount,                                                                                                    "+
                        "	ROW_NUMBER() OVER(PARTITION BY window_start,window_end,product_brand ORDER BY real_amount DESC) as rn            "+
                        "FROM o1 ) o2                                                                                                        "+
                        "WHERE rn<=2                                                                                                         "
        ).print();



        env.execute();


    }

}
